package bsp.scan;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.FileInputFormat;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;

import util.DiffJobA;
import util.functors.SumLong;

public class Scan extends BspScanHelper {
	public static final Log LOG = LogFactory.getLog(Scan.class);

	public static DiffJobA createScanJob( final Path in, final Path out) throws IOException{
		HamaConfiguration conf = new HamaConfiguration();
		DiffJobA job = new DiffJobA(conf);
		job.setInputFormat(TextInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		job.setOutputFormat(TextOutputFormat.class);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
//		BSPJobClient jobClient = new BSPJobClient(job.getConfiguration());
//		ClusterStatus cluster = jobClient.getClusterStatus(true);
		return job;
	}
	
	@Override
	public void setOperator() {
		this.oplus = new SumLong();
	}
}